#include <slothttpd/aio/aio_operation.hpp>
#include <slothttpd/logger.hpp>

#include <stdexcept>
#include <iostream>

#include <sys/eventfd.h>
#include <sys/stat.h>
#include <fcntl.h>

#include <boost/throw_exception.hpp>
#include <boost/bind.hpp>

namespace slothttpd {
namespace aio {

base_async_operation::base_async_operation() :
    io_ctx(nullptr)
{
}

bool base_async_operation::cancel()
{
    if (is_pending()) {
        return io_ctx->cancel(*this);
    } else {
        return true;
    }
}

void base_async_operation::complete(std::size_t length, const boost::system::error_code &err)
{
    io_ctx = nullptr;
    try {
        on_complete_handler(length, err);
        on_complete_handler = operation_handler();
    } catch (...) {
        on_complete_handler = operation_handler();
        throw;
    }
}

void base_async_operation::wait()
{
    if (is_pending()) {
        io_ctx->wait_for(this);
    }
}

bool base_async_operation::is_pending() const
{
    return io_ctx != nullptr;
}

base_async_operation::~base_async_operation()
{
    if (is_pending() && !cancel()) {
        // operation in progress. TODO: wait until completed
        // This code should not be executed, you should wait for op completion before destroying object
        logger() << "waiting for async op in destructor" << std::endl;
        
        io_ctx->wait_for(this);
    }
}

void base_async_operation::submit_cmd(linux_aio_context &ctx, file_ptr f, char *buffer, std::size_t size, off64_t off, 
                                      operation_handler &handler, int cmd)
{
    if (!is_pending()) {
        {
            __u64 b = reinterpret_cast<__u64>(buffer);
            if (b % 512 != 0) {
                logger() << "base_async_operation::submit_cmd(): buffer is not aligned to 512b" << std::endl;
            }
        }
        
        std::memset(&io_block, 0, sizeof io_block);
        on_complete_handler = handler;
        
        io_block.aio_fildes     = f->descriptor();
        io_block.aio_offset     = off;
        io_block.aio_buf        = reinterpret_cast<__u64>(buffer);
        io_block.aio_nbytes     = size;
        io_block.aio_lio_opcode = cmd;
        io_block.aio_flags      = IOCB_FLAG_RESFD;
        io_block.aio_resfd      = ctx.eventfd_stream_.native_handle();
        io_block.aio_data       = reinterpret_cast<__u64>(this);
        
        if (ctx.submit(*this)) {
            io_ctx = &ctx;
        }
    } else {
        BOOST_THROW_EXCEPTION(std::logic_error("operation already pending"));
    }
}

void async_read::submit(linux_aio_context &ctx, file_ptr f, char *buffer, std::size_t size,
                        off64_t off, operation_handler &handler)
{
    return submit_cmd(ctx, f, buffer, size, off, handler, IOCB_CMD_PREAD);
}

void async_write::submit(linux_aio_context &ctx, file_ptr f, const char *buffer, std::size_t size,
                         off64_t off, operation_handler &handler)
{
    return submit_cmd(ctx, f, const_cast<char*>(buffer), size, off, handler, IOCB_CMD_PWRITE);
}

const unsigned linux_aio_context::default_number_of_events = 128;

linux_aio_context::linux_aio_context(boost::asio::io_service &ios, unsigned nr_events) :
    ctx_id(0),   // should be filled with 0 before its address passed to 'io_setup'
    ios_(ios),
    eventfd_stream_(ios),
    eventfd_stream_read_pending_(false)
{
    int completion_fd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (completion_fd <= 0) {
        BOOST_THROW_EXCEPTION(std::runtime_error(strerror(errno)));
    }
    
    try {
        eventfd_stream_.assign(completion_fd);
    } catch (...) {
        ::close(completion_fd);
        throw;
    }
    
    if (::io_setup(nr_events, &ctx_id) != 0) {
        BOOST_THROW_EXCEPTION(std::runtime_error(strerror(errno)));
    }
}

linux_aio_context::~linux_aio_context() noexcept
{
    wait_all();
    
    if (::io_destroy(ctx_id) != 0) {
        logger() << "cannot destroy aio_context_t: " << strerror(errno) << std::endl;
    }
}

void linux_aio_context::wait_for(base_async_operation &wait_op)
{
    return wait_for(&wait_op);
}


void linux_aio_context::wait_all()
{
    wait_for(nullptr);
}

void linux_aio_context::cancel_all()
{
    for (auto it = ops.begin(); it != ops.end(); ) {
        base_async_operation &op = *it;
        
        // increment here because 'cancel' can unlink node.
        // TOOD: check if we can use collection 'for' here
        ++it;
        op.cancel();
    }
}

bool linux_aio_context::submit(base_async_operation &op)
{
    struct iocb *iocbp[1] = { &op.io_block };
    
    if (::io_submit(ctx_id, 1, iocbp) > 0) {
        ops.push_front(op);
        post_read_eventfd();
        return true;
    } else {
        logger() << "io_submit(): " << strerror(errno) << '\n';
        
        boost::system::error_code error(errno, boost::system::system_category());
        op.complete(0, error);
        
        return false;
    }
}

bool linux_aio_context::cancel(base_async_operation &op)
{
    struct io_event ev;
    
    if (::io_cancel(ctx_id, &op.io_block, &ev) == 0) {
        auto error = boost::asio::error::operation_aborted;
        
        op.unlink();
        op.complete(static_cast<std::size_t>(ev.res), error);
        
        return true;
    } else if (errno != EAGAIN) {
        // TODO: why this returns error ?
        logger() << "io_cancel(): " << strerror(errno) << '\n';
    }
    
    return false;
}

void linux_aio_context::process_operations()
{
    struct io_event events[default_number_of_events];
    int got_events;
    
    while ((got_events = ::io_getevents(ctx_id, 0, default_number_of_events, events, NULL)) > 0) {
        for (int i = 0; i < got_events; i++) {
            complete_op(events[i]);
        }
    }
    
    if (got_events < 0) {
        logger() << "io_getevents(): " << strerror(errno) << '\n';
    }
}

bool linux_aio_context::has_operations_pending() const
{
    return !ops.empty();
}

void linux_aio_context::complete_op(struct io_event &ev)
{
    auto op = reinterpret_cast<base_async_operation*>(ev.data);
    std::size_t bytes = (ev.res < 0) ? 0 : static_cast<std::size_t>(ev.res);
    int op_errno = (ev.res < 0) ? -ev.res : 0;
    
    op->unlink();
    
    boost::system::error_code err(op_errno, boost::system::system_category());
    
    if (op_errno == 0 && bytes == 0)
        err = boost::asio::error::eof;
    op->complete(bytes, err);
}

void linux_aio_context::wait_for(base_async_operation *wait_this)
{
    const int nr_events = 128;
    struct io_event events[nr_events];
    
    bool wait_complete = (wait_this == nullptr);
    
    while (!ops.empty() || !wait_complete) {
        int got_events = ::io_getevents(ctx_id, 0, nr_events, events, NULL);
        
        // TODO: remove this code. I put it here just to check if I do everything right
        if (got_events == 0) {
            logger() << "there are async ops submitted, but io_getevents returns 0!\n";
        }
        
        for (int i = 0; i < got_events; i++) {
            complete_op(events[i]);
            
            if (wait_this == reinterpret_cast<base_async_operation*>(events[i].data)) {
                wait_complete = true;
            }
        }
    }
}

void linux_aio_context::stop()
{
    eventfd_stream_.cancel();
}

void linux_aio_context::post_read_eventfd()
{
    if (!eventfd_stream_read_pending_) {
        eventfd_stream_.async_read_some(boost::asio::buffer(event_buffer_, sizeof event_buffer_),
                                        boost::bind(&linux_aio_context::handle_read_eventfd,
                                                    this,
                                                    boost::asio::placeholders::bytes_transferred,
                                                    boost::asio::placeholders::error));
        eventfd_stream_read_pending_ = true;
    }
}

void linux_aio_context::handle_read_eventfd(std::size_t bytes_transferred, const boost::system::error_code &err)
{
    eventfd_stream_read_pending_ = false;
    
    (void) bytes_transferred;
    
    if (err) {
        logger() << err << '\n';
    }
    
    process_operations();
    
    if (has_operations_pending()) {
        post_read_eventfd();
    }
}

}
}
